package org.fujene.communicate.indexer;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.zip.GZIPInputStream;
import java.util.zip.GZIPOutputStream;

import org.fujene.constants.Constants;
import org.fujene.logging.LogParser;
import org.fujene.memman.ReplicateMemoryManager;
import org.fujene.structure.LookupPointer;
import org.fujene.structure.Queue;
import org.fujene.toolbox.ByteConverter;

public class ContentIndexerClient implements Runnable{
	private static final int STRLEN = Constants.BYTE_STRLEN;
	private static ReplicateMemoryManager memman;
	private static LinkedBlockingQueue<Queue> clientQueue;
	private static Object clientBlockade;
	private static GZIPOutputStream clientLog;
	private static File logFile;

	static{
		clientQueue = new LinkedBlockingQueue<Queue>();
		clientBlockade = new Object();
	}

	public ContentIndexerClient(boolean isCreateLog){
		if(isCreateLog && logFile == null){
			logFile = new File(".content_clientlog");
			if(!logFile.exists()) try{
				logFile.createNewFile();
			}catch(IOException e){
				System.err.println("Client-log file does not create. Indexing works without logging.");
			}
			try{
				clientLog = new GZIPOutputStream(new FileOutputStream(logFile));
			}catch(FileNotFoundException e){}catch(IOException e){}
		}
	}

	public static void property(ReplicateMemoryManager manager){
		memman = manager;
	}

	private static void addLog(Queue queue) throws IOException{
		byte[] strbytes;
		clientLog.write(ByteConverter.longToByte(queue.ID, 8));
		clientLog.write(ByteConverter.intToByte(queue.contentTypes, 1));
		clientLog.write(ByteConverter.intToByte(queue.appendixTypes, 1));
		for(int i = 0; i < queue.contentTypes; i++){
			strbytes = queue.contents[i].getBytes();
			clientLog.write(ByteConverter.intToByte(strbytes.length, STRLEN));
			clientLog.write(strbytes);
		}
		for(int i = 0; i < queue.appendixTypes; i++){
			strbytes = queue.appendices[i].getBytes();
			clientLog.write(ByteConverter.intToByte(strbytes.length, STRLEN));
			clientLog.write(strbytes);
		}
	}

	// Clear log file and re-construct
	public static void clearLog() throws IOException{
		if(clientLog != null){
			clientLog.close();
			logFile.delete();
			logFile.createNewFile();
			clientLog = new GZIPOutputStream(new FileOutputStream(logFile));
		}
	}

	// Revert from log file
	public static void revert() throws FileNotFoundException, IOException{
		GZIPInputStream logstream = null;
		try{
			logstream = new GZIPInputStream(new FileInputStream(logFile));
			byte[] strlenb = new byte[STRLEN];
			byte[] strb;
			long ID;
			int contentType, appendixType;
			Queue queue;
			while(logstream.available() != 0){
				strb = new byte[8];
				logstream.read(strb);
				ID = ByteConverter.byteToLong(strb);
				strb = new byte[1];
				logstream.read(strb);
				contentType = ByteConverter.byteToInt(strb);
				logstream.read(strb);
				appendixType = ByteConverter.byteToInt(strb);

				queue = new Queue(contentType, appendixType, ID);
				for(int i = 0; i < contentType; i++){
					logstream.read(strlenb);
					strb = new byte[ByteConverter.byteToInt(strlenb)];
					logstream.read(strb);
					queue.contents[i] = new String(strb, "UTF-8");
				}
				for(int i = 0; i < appendixType; i++){
					logstream.read(strlenb);
					strb = new byte[ByteConverter.byteToInt(strlenb)];
					logstream.read(strb);
					queue.appendices[i] = new String(strb, "UTF-8");
				}
				clientQueue.add(queue);
			}
		}finally{
			if(logstream != null) logstream.close();
		}
	}

	public static void addToQueue(Queue newIndex){
		// FIXME: Add log
		if(clientLog != null) try{
			addLog(newIndex);
		}catch(IOException e){
			e.printStackTrace();
		}
		clientQueue.offer(newIndex);
	}

	public static boolean isEmpty(){
		return clientQueue.isEmpty();
	}

	public static void waitUntilEmpty(){
		if(!clientQueue.isEmpty()) try{
			clientBlockade.wait();
		}catch(InterruptedException e){}
	}

	public static int residue(){
		return clientQueue.size();
	}

	@Override
	public void run(){
		Queue newIndex = null;
		while(true){
			// Take from client queue
			try{
				newIndex = clientQueue.take();
				if(clientQueue.isEmpty()) synchronized(clientBlockade){
					clientBlockade.notifyAll();
				}
			}catch(InterruptedException e1){
				continue;
			}
			long startTime = LogParser.initTime();
			LookupPointer[] appPtr = new LookupPointer[newIndex.appendixTypes];
			for(int i = 0; i < newIndex.appendixTypes; i++)
				appPtr[i] = memman.storeAppendix(newIndex.appendices[i], newIndex.ID, i);
			startTime = LogParser.timerFinest("Index / Index appendix", startTime);
			for(int i = 0; i < newIndex.contentTypes; i++)
				memman.storeContent(newIndex.contents[i], newIndex.ID, i, appPtr);
			startTime = LogParser.timerFinest("Index / Index content", startTime);
		}
	}

	@Override
	protected void finalize() throws Throwable{
		if(clientLog != null) clientLog.close();
		super.finalize();
	}
}
